/*
 * StorageChannelThread.cpp
 *
 *  Created on: 2017年7月5日
 *      Author: donna
 */

#include "StorageChannelThread.h"

namespace av
{

StorageChannelThread::StorageChannelThread():
	m_sink()
{
}

StorageChannelThread::~StorageChannelThread()
{
	close();
}

bool StorageChannelThread::open()
{
	return start();
}

void StorageChannelThread::close()
{
	if (isRunning())
	{
		stop();
	}
}

void StorageChannelThread::setSink(MediaStorageSink* sink)
{
	m_sink = sink;
}

void StorageChannelThread::setDataStore(DataStorePtr& dataStore)
{
	m_dataStore = dataStore;
}

void StorageChannelThread::postStorageEvent(StorageEvent& storageEvent)
{
	{
		comn::AutoCritSec lock(m_cs);
		m_events.push_back(storageEvent);
	}
	m_event.post();
}

void StorageChannelThread::postRecord(MRecord& record, bool isAdd)
{
	RecordEntry entry;
	memcpy(&entry.record, &record, sizeof(record));
	entry.isAdd = isAdd;

	{
		comn::AutoCritSec lock(m_cs);
		m_records.push_back(entry);
	}
	m_event.post();
}

void StorageChannelThread::postWriter(MediaWriterPtr& writer)
{
	if (!writer)
	{
		return;
	}

	{
		comn::AutoCritSec lock(m_cs);
		m_writers.push_back(writer);
	}
	m_event.post();
}

int StorageChannelThread::run()
{
	while (!m_canExit)
	{
		m_event.timedwait(-1);

		handleWriter();
		handleRecord();
		handleStorageEvent();

		if (m_canExit)
		{
			break;
		}
	}
	return 0;
}

void StorageChannelThread::doStop()
{
	m_event.post();
}

void StorageChannelThread::handleWriter()
{
	while (true)
	{
		MediaWriterPtr writer;
		{
			comn::AutoCritSec lock(m_cs);
			if (m_writers.size())
			{
				writer = m_writers.front();
				m_writers.pop_front();
			}
			else
			{
				break;
			}
		}

		if (!writer)
		{
			continue;
		}

		writer->close();
	}
}

void StorageChannelThread::handleRecord()
{
	while (true)
	{
		RecordEntry recordEntry;
		{
			comn::AutoCritSec lock(m_cs);
			if (m_records.size())
			{
				recordEntry = m_records.front();
				m_records.pop_front();
			}
			else
			{
				break;
			}
		}

		if (recordEntry.isAdd)
		{
			m_dataStore->addRecord(recordEntry.record);
		}
		else
		{
			m_dataStore->updateRecord(recordEntry.record);
		}
	}
}

void StorageChannelThread::handleStorageEvent()
{
	while (true)
	{
		StorageEvent storageEvent;
		{
			comn::AutoCritSec lock(m_cs);
			if (m_events.size())
			{
				storageEvent = m_events.front();
				m_events.pop_front();
			}
			else
			{
				break;
			}
		}

		if (m_sink)
		{
			m_sink->onStorageEvent(storageEvent);
		}
	}
}


} /* namespace av */
